package com.yida.gwj.expend.handler;

import java.util.Optional;
import java.util.concurrent.TimeUnit;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.yida.common.core.exception.ServiceException;
import com.yida.common.redis.service.RedisService;
import com.yida.gwj.api.service.RemoteGwjConfigService;
import com.yida.gwj.common.enums.GwjBusinessFlagType;
import com.yida.gwj.common.utils.GwjEnumUtils;
import com.yida.gwj.expend.config.MaintenanceFundsBusinessConfiguration;
import com.yida.gwj.expend.domain.message.MqBaseResult;
import com.yida.gwj.expend.enums.MaintenanceFundsErrorEnum;
import com.yida.gwj.expend.service.IGwjMaintenanceFundsBusinessAttachmentService;
import com.yida.gwj.expend.service.IGwjMaintenanceFundsBusinessLogService;
import com.yida.gwj.expend.service.IGwjMaintenanceFundsBusinessService;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;

/**
 * 消费
 *
 * @author liangp
 * @date 2023/03/01
 */
@Component
@RocketMQMessageListener(topic = "${gwj.expend.topic}", consumerGroup = "${gwj.expend.group.pre-inner}", selectorExpression = "${gwj.expend.tag.pre-accept-submit} || ${gwj.expend.tag.pre-accept-second-submit}", messageModel = MessageModel.CLUSTERING)
public class ExpendInnerListenerHandler implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    private final Logger log = LoggerFactory.getLogger(ExpendInnerListenerHandler.class);

    private final String cache = "gwj:" + MaintenanceFundsBusinessConfiguration.maintenanceFundsBusinessTopic + ":mq:";

    @Autowired
    private RedisService redisService;
    @Autowired
    private MaintenanceFundsBusinessConfiguration maintenanceFundsBusinessConfiguration;

    private final HandlerExpendPreService maintenanceFundsPreServiceHandler;

    public ExpendInnerListenerHandler(RemoteGwjConfigService remoteGwjConfigService, IGwjMaintenanceFundsBusinessService maintenanceFundsBusinessService, IGwjMaintenanceFundsBusinessLogService maintenanceFundsBusinessLogService, IGwjMaintenanceFundsBusinessAttachmentService maintenanceFundsBusinessAttachmentService) {
        this.maintenanceFundsPreServiceHandler = new HandlerExpendPreService(remoteGwjConfigService, maintenanceFundsBusinessService, maintenanceFundsBusinessLogService, maintenanceFundsBusinessAttachmentService);
    }

    /**
     * 消息接收处理方法
     *
     * @param message 消息
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void onMessage(MessageExt message) {
        log.info("[MQ][INNER] topic -> [{}], tag -> [{}] 消息开始处理.....", message.getTopic(), message.getTags());
        TimeInterval timer = null;
        if (log.isDebugEnabled()) {
            timer = DateUtil.timer();
            log.debug("[MQ][INNER] 参数体 -> [{}]", message);
        }
        try {
            if (redisService.hasKey(cache + message.getKeys())) {
                log.error("当前消息已经消费,key值 -> [{}]", message.getKeys());
                return;
            }
            Optional<JSONObject> optionalJson = JSONUtil.toBean(StrUtil.utf8Str(message.getBody()), Optional.class);
            optionalJson.<ServiceException>orElseThrow(() -> {
                throw new ServiceException(GwjEnumUtils.getEnumConstantMsg(MaintenanceFundsErrorEnum.MQ_MAINTENANCE_FUNDS_INNER_LISTENER) + "MQ基础信息获取失败");
            });
            MqBaseResult mqBaseResult = JSONUtil.toBean(optionalJson.get(), MqBaseResult.class);
            if (ObjectUtil.equal(message.getTags(), maintenanceFundsBusinessConfiguration.tagProperties().getPreAcceptSubmit())) {
                maintenanceFundsPreServiceHandler.preAcceptSubmit(mqBaseResult, GwjBusinessFlagType.EXPEND_PRE);
            }
            if (ObjectUtil.equal(message.getTags(), maintenanceFundsBusinessConfiguration.tagProperties().getPreAcceptSecondSubmit())) {
                maintenanceFundsPreServiceHandler.preAcceptSecondSubmit(mqBaseResult);
            }
            redisService.setCacheObject(cache + message.getKeys(), message.getKeys(), 1L, TimeUnit.DAYS);
            if (log.isDebugEnabled()) {
                log.debug("消费处理时间 [{}] 毫秒", timer.interval());
            }
            log.info("[MQ][INNER] topic -> [{}], tag -> [{}] 消息处理完成", message.getTopic(), message.getTags());
        } catch (Exception e) {
            log.error("[MQ][InnerListenerHandler]公维金资金申请内网处理异常 -> [{}]", e);
            throw new ServiceException(GwjEnumUtils.getEnumConstantMsg(MaintenanceFundsErrorEnum.MQ_MAINTENANCE_FUNDS_INNER_LISTENER) + e.getMessage());
        }
    }

    /**
     * 消息接受前置处理
     *
     * @param defaultMqPushConsumer
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMqPushConsumer) {
        defaultMqPushConsumer.setMaxReconsumeTimes(3);
    }

}
